package GraghX

import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object graphX_demo {
  def main(args: Array[String]): Unit = {

    //设置运行环境
    val conf = new SparkConf().setAppName("graphX_demo").setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    //设置users顶点
    val users: RDD[(VertexId, (String, Int))] =
      sc.parallelize(Array((1L, ("Alice", 28)),
        (2L, ("Bob", 27)),
        (3L, ("Charlie", 65)),
        (4L, ("David", 42)),
        (5L, ("Ed", 55)),
        (6L, ("Fran", 50))
      ))

    //设置relationships边
    val relationships: RDD[Edge[Int]] =
      sc.parallelize(Array  (
        Edge(2L, 1L, 7),
        Edge(2L, 4L, 2),
        Edge(3L, 2L, 4),
        Edge(3L, 6L, 3),
        Edge(4L, 1L, 1),
        Edge(5L, 2L, 2),
        Edge(5L, 3L, 8),
        Edge(5L, 6L, 3)
      ))


    // Build the initial Graph
    val graph = Graph(users, relationships)

    println("---------------------------------------------")


    /*println("找到图中属性是student的顶点")
    graph.vertices.filter { case (id, (name, occupation)) => occupation == "student" }.collect.foreach {
      case (id, (name, occupation)) => println(s"$name is $occupation")
    }

    println("---------------------------------------------")
    println("找到图中边属性是advisor的边")
    graph.edges.filter(e => e.attr == "advisor").collect.foreach(e => println(s"${e.srcId} to ${e.dstId} att ${e.attr}"))*/

    // println("---------------------------------------------")
   // println("找出图中最大的出度、入度、度数：")
  //  def max(a: (VertexId, Int), b: (VertexId, Int)): (VertexId, Int) = {
 //     if (a._2 > b._2) a else b
   // }
    //println("****************************************")
    //var tuples: Array[(VertexId, Int)] = graph.outDegrees.collect()
    //tuples.foreach(println(_))

    //graph.edges.collect.foreach(println(_))
    //println("****************************************")

 //   println("max of outDegrees:" + graph.outDegrees.reduce(max) + " max of inDegrees:" + graph.inDegrees.reduce(max) + " max of Degrees:" + graph.degrees.reduce(max))







    //构造图 Graph[VD,ED]
   // val graph: Graph[(String, Int), Int] = Graph(users, relationships)
    println("***********************************************")
    println("属性演示")
    println("**********************************************************")
    println("找出图中年龄大于 30 的顶点：")
    graph.vertices.filter { case (id, (name, age)) => age > 30 }.collect.foreach {
      case (id, (name, age)) => println(s"$name is $age")
    }
    graph.triplets.foreach(t => println(s"triplet:${t.srcId},${t.srcAttr},${t.dstId},${t.dstAttr},${t.attr}"))
    //边操作：找出图中属性大于 5 的边
    println("找出图中属性大于 5 的边：")
    graph.edges.filter(e => e.attr > 5).collect.foreach(e => println(s"${e.srcId} to ${e.dstId} att ${e.attr}"))
    println
    //triplets 操作，((srcId, srcAttr), (dstId, dstAttr), attr)
    println("列出边属性>5 的 tripltes：")
    for (triplet <- graph.triplets.filter(t => t.attr > 5).collect) {
      println(s"${triplet.srcAttr._1} likes ${triplet.dstAttr._1}")
    }
    println
    //Degrees 操作
    println("找出图中最大的出度、入度、度数：")

    def max(a: (VertexId, Int), b: (VertexId, Int)): (VertexId, Int) = {
      if (a._2 > b._2) a else b
    }

    println("max of outDegrees:" + graph.outDegrees.reduce(max) + ", max of inDegrees:" + graph.inDegrees.reduce(max) + ", max of Degrees:" +
      graph.degrees.reduce(max))




  }

}
